package edu.zju.gis.dbfg.model.util;

import com.google.gson.Gson;
import org.locationtech.jts.geom.Geometry;
import org.locationtech.jts.io.WKTReader;
//import com.vividsolutions.jts.geom.Geometry;
//import com.vividsolutions.jts.io.WKTReader;
import edu.zju.gis.dbfg.common.exception.LoaderException;
import edu.zju.gis.dbfg.model.common.StringOutputStream;
import edu.zju.gis.dbfg.model.entity.ConstructedDataField;
import edu.zju.gis.dbfg.model.entity.DataType;
import edu.zju.gis.dbfg.model.entity.FeatureType;
import edu.zju.gis.dbfg.model.exception.ModelElasticsearchException;
import edu.zju.gis.dbfg.model.exception.ModelFailedException;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
//import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse;
//import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequestBuilder;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
//import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;

import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.geotools.geojson.geom.GeometryJSON;
import org.json.JSONException;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.io.IOException;
import java.io.Serializable;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;

/**
 * @author Hu
 * @date 2020/1/2
 **/
public class EsHelper implements Serializable {

    private final static Logger logger = LoggerFactory.getLogger(EsHelper.class);
    public static String NUMBER_OF_SHARDS_NAME = "index.number_of_shards";
    public static String NUMBER_OF_REPLICA_NAME = "index.number_of_replicas";
    public static String REFRESH_INTERVAL_NAME = "index.refresh_interval";
    public static int RETRY_TIME = 3;
    protected static int INDEX_TIME_OUT = 200; // time unit is minute
    protected static int NUMBER_OF_SHARDS = 3;
    protected static int NUMBER_OF_REPLICA = 0;
    protected static int INDEX_UNREFRESH_INTERVAL = -1;
    protected static int INDEX_REFRESH_INTERVAL = 2;
    // TODO put these configuration into property file
    private static String ES_CLUSTER_NAME = "my-application";
    private static String DEFAULT_TYPE_NAME = "doc";
    //    private static String[] ES_HOST_IP = new String[]{"nn01", "nn02", "dn01", "dn02", "dn07", "dn04", "dn05", "dn06"};
    private static String[] ES_HOST_IP = new String[]{"121.89.211.142"};
    private static int ES_TRANSPORT_PORT = 9300;
    private static int ES_HTTP_PORT = 9200;
    private static EsHelper esHelper;

    transient private TransportClient esClient;

    public EsHelper() {

//        this.esClient = initEsClient();
        try {
            System.setProperty("es.set.netty.runtime.available.processors", "false");
            Settings settings = Settings.builder()
                    .put("cluster.name", ES_CLUSTER_NAME).build();
//            Settings settings = Settings.builder()
//                    .put("cluster.name", ES_CLUSTER_NAME)
//                    .put("thread_pool.search.size", 2)
////                    .put("thread_pool.bulk.size", 2)
////                    .put("client.transport.sniff", true)
////                    .put("transport.netty.worker_count", 1)
//                    .build();
            this.esClient = new PreBuiltTransportClient(settings).addTransportAddress(new TransportAddress(InetAddress.getByName(ES_HOST_IP[0]), ES_TRANSPORT_PORT));
        } catch (UnknownHostException e) {
            this.finish();
            try {
                throw e;
            } catch (UnknownHostException e1) {
                e1.printStackTrace();
            }
        }
    }

    public static synchronized EsHelper getInstanceClient() {
        if (esHelper == null) {
            esHelper = new EsHelper();
        }
        return esHelper;
    }

//    private TransportClient initEsClient() {
//        try {
//            // TODO 根据 spark executor-num 以及 executor-cores 动态调整线程数
//            Settings settings = Settings.builder()
//                    .put("cluster.name", ES_CLUSTER_NAME)
//                    .put("thread_pool.search.size", 2)
//                    .put("thread_pool.bulk.size", 2)
//                    .put("client.transport.sniff", true)
//                    .put("transport.netty.worker_count", 1)
//                    .build();
//
//            TransportClient client = new PreBuiltTransportClient(settings).addTransportAddress(new TransportAddress(InetAddress.getByName(ES_HOST_IP[0]), ES_TRANSPORT_PORT));
//
////            for (String host : ES_HOST_IP) {
////                try {
////                    client.addTransportAddress(new TransportAddress(InetAddress.getByName(host), ES_TRANSPORT_PORT));
////                } catch (UnknownHostException e) {
////                    throw new ModelElasticsearchException(EsHelper.class, "initEsClient()", "Init es client failed: " + e.getMessage(), host);
////                }
////            }
//            return client;
//        } catch (Exception e) {
//           this.finish();
//            try {
//                throw e;
//            } catch (UnknownHostException e1) {
//                e1.printStackTrace();
//            }
//        }
//        return;
//    }

    /**
     * 获取 source 的字段元数据信息（mapping）
     *
     * @param source
     * @return
     */
    public Map<String, Object> getEsSchema(String source) {
        String index = source.split("/")[0];
        String type = source.split("/")[1];
        ClusterState cs = this.esClient.admin().cluster().prepareState().setIndices(index).execute().actionGet().getState();
        IndexMetaData imd = cs.getMetaData().index(index);
//        MappingMetaData mdd = imd.mapping(type);
        MappingMetaData mdd = imd.mapping();
        try {
            return (Map<String, Object>) mdd.getSourceAsMap().getOrDefault("properties", new HashMap<>());
        } catch (Exception e) {
            throw new ModelElasticsearchException(EsHelper.class, "getEsSchema()", "get es metadata faield", source);
        }
    }

    public boolean indexExists(String index) {
        return this.esClient.admin().indices().prepareExists(index).get().isExists();
    }

    public boolean typeExists(String index, String... type) {
        return this.esClient.admin().indices().prepareTypesExists(index).setTypes(type).get().isExists();
    }

    public boolean createIndex(String index, String type, String mapping) {
        if (this.indexExists(index) && this.typeExists(index, type)) {
            System.out.println("已经存在索引");
            this.esClient.admin().indices().prepareDelete(index).execute().actionGet();
            System.out.println("删除已存在索引");
//            throw new ModelFailedException(EsHelper.class, "createIndex()", "already exists", String.format("%s/%s", index, type));
        }
        CreateIndexResponse response = this.esClient.admin().indices().prepareCreate(index)
                .setSettings(Settings.builder()
                        .put(NUMBER_OF_SHARDS_NAME, NUMBER_OF_SHARDS)
                        .put(NUMBER_OF_REPLICA_NAME, NUMBER_OF_REPLICA)
                ).addMapping(type, mapping, XContentType.JSON).get();
        StringOutputStream sos = new StringOutputStream();
        try {
            response.writeTo(sos);
            logger.info(sos.getString());
            System.out.println("索引构建成功");
        } catch (IOException e) {
            e.printStackTrace();
        }
        return response.isAcknowledged();
    }

    public boolean createIndex(String index, String type) {
        return this.createIndex(index, type, "{}");
    }

    public boolean createIndex(String index) {
        return this.createIndex(index, DEFAULT_TYPE_NAME, "{}");
    }

    public boolean stopRefresh(String index) {
        Settings setting = Settings.builder()
                .put(REFRESH_INTERVAL_NAME, String.format("%d", INDEX_UNREFRESH_INTERVAL)).build();
        return this.updateIndexSettings(index, setting);
    }

    public boolean startRefresh(String index) {
        return this.startRefresh(index, INDEX_REFRESH_INTERVAL);
    }

    public boolean startRefresh(String index, int interval) {
        Settings setting = Settings.builder()
                .put(REFRESH_INTERVAL_NAME, String.format("%ds", interval)).build();
        return this.updateIndexSettings(index, setting);
    }

    public boolean updateIndexSettings(String index, Settings settings) {
//        UpdateSettingsResponse response = this.esClient.admin().indices().prepareUpdateSettings(index)
//                .setSettings(settings)
//                .get();
        return this.esClient.admin().indices().prepareUpdateSettings(index)
                .setSettings(settings)
                .get().isAcknowledged();
    }

    public boolean updateIndexMapping(String index, String type, String mapping) {
//        PutMappingResponse response = this.esClient.admin().indices().preparePutMapping(index)
//                .setType(type)
//                .setSource(mapping, XContentType.JSON)
//                .get();
        return this.esClient.admin().indices().preparePutMapping(index)
                .setType(type)
                .setSource(mapping, XContentType.JSON)
                .get().isAcknowledged();
    }

    public boolean updateIndexMapping(String index, String mapping) {
        return this.updateIndexMapping(index, DEFAULT_TYPE_NAME, mapping);
    }

    /**
     * @param index 索引 id
     * @param type  索引 type
     * @param ts    _1: id, _2: value in json format
     * @return 将错误的tuple数据id以及错误message返回
     */
    public Map<String, String> batchLoad(String index, String type, List<Tuple2<String, String>> ts) {

        Map<String, String> errors = new HashMap<>();

        BulkRequestBuilder bulkRequest = this.esClient.prepareBulk();

        bulkRequest.setTimeout(TimeValue.timeValueMinutes(INDEX_TIME_OUT));

        for (Tuple2<String, String> t : ts) {
            String id = t._1;
            String json = t._2;
            bulkRequest.add(this.esClient.prepareIndex(index, type, id)
                    .setSource(json, XContentType.JSON));
        }
//        System.out.println("ts的长度是"+ts.size());
        System.out.println("bulkRequest" + bulkRequest.toString());
        if (bulkRequest.numberOfActions() > 0) {
            BulkResponse bulkResponse = bulkRequest.get();
            if (bulkResponse.hasFailures()) {
                BulkItemResponse[] responses = bulkResponse.getItems();
                for (int i = 0; i < responses.length; i++) {
                    BulkItemResponse response = responses[i];
                    if (response.isFailed()) {
                        StringBuilder sb = new StringBuilder();
                        sb.append("\nLoad Error:[")
                                .append(i)
                                .append("]: index [")
                                .append(response.getIndex())
                                .append("], type [")
                                .append(response.getType())
                                .append("], id [")
                                .append(response.getId())
                                .append("], message [")
                                .append(response.getFailureMessage())
                                .append("]");
                        errors.put(response.getId(), sb.toString());
//                    logger.info()
                        System.out.println(sb.toString());
                    }
                }
            }

        }
        return errors;
    }


    public void finish() {
        if (this.esClient != null) {
            this.esClient.close();
        }
    }

}
